package samples

import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.ExperimentalCoroutinesApi
import kotlinx.coroutines.FlowPreview
import kotlinx.coroutines.cancel
import kotlinx.coroutines.currentCoroutineContext
import kotlinx.coroutines.delay
import kotlinx.coroutines.ensureActive
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.asFlow
import kotlinx.coroutines.flow.buffer
import kotlinx.coroutines.flow.cancellable
import kotlinx.coroutines.flow.catch
import kotlinx.coroutines.flow.collect
import kotlinx.coroutines.flow.collectLatest
import kotlinx.coroutines.flow.combine
import kotlinx.coroutines.flow.conflate
import kotlinx.coroutines.flow.flatMapConcat
import kotlinx.coroutines.flow.flatMapLatest
import kotlinx.coroutines.flow.flatMapMerge
import kotlinx.coroutines.flow.flattenConcat
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.flow.flowOf
import kotlinx.coroutines.flow.flowOn
import kotlinx.coroutines.flow.launchIn
import kotlinx.coroutines.flow.map
import kotlinx.coroutines.flow.onCompletion
import kotlinx.coroutines.flow.onEach
import kotlinx.coroutines.flow.reduce
import kotlinx.coroutines.flow.transform
import kotlinx.coroutines.flow.zip
import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.test.runTest
import org.junit.Test
import kotlin.system.measureTimeMillis

/**
 * Flow测试
 * collect这种后面没有任何返回值的叫做Terminal Operator；
 * map、filter这种叫做Intermediate Operator；
 * @author mph
 * @date 2025/1/10
 */
@OptIn(ExperimentalCoroutinesApi::class)
class FlowTest {

    @Test
    fun test1() = runTest {
        //FLow是顺序执行，就是每个元素沿着调用链执行一遍，而不是所有元素都执行完才执行下一步方法，比如下面代码输出：
        // map 1
        // transform 1
        // collect 1
        // map 2
        // transform 4
        // collect 4
        // map 3
        // transform 9
        // collect 9
        (1..3).asFlow()
            .map {
                log("map $it")
                it * it
            }
            .transform {
                log("transform $it")
                emit(it)
            }
            .collect {
                log("collect $it")
            }
        val result = (1..5).reduce { sum, value -> sum + value } //reduce返回值是具体的值，而不是Flow类型，意味着不能再链式调用
        log(result)

//        simpleWrong().collect { value -> println(value) }
//        simpleFlowOn().collect { value -> println(value) }

        //因为Flow的默认机制是顺序执行，因此这里的耗时会更长（大概100*3+300*3）
//        val time = measureTimeMillis {
//            simpleComputation().collect { value ->
//                delay(300) // pretend we are processing it for 300 ms
//                println(value)
//            }
//        }
        //注意，flowOn的机制其实就是使用的buffer，不过这里不需要切换执行线程
//        val time = measureTimeMillis {
//            simpleComputation()
//                .buffer() //加上buffer后耗时是单次循环simpleComputation()执行时间加上collect的总时间（大概是100+300*3）
//                .collect { value ->
//                    delay(300) // pretend we are processing it for 300 ms
//                    println(value)
//                }
//        }
//        println("Collected in $time ms")
    }

    /**
     * Flow的默认机制是flow内部的执行线程和collect的执行线程是同一个线程，如果强行通过指定Dispatcher的方式会抛出异常
     */
    private fun simpleWrong(): Flow<Int> = flow {
        // The WRONG way to change context for CPU-consuming code in flow builder
        kotlinx.coroutines.withContext(Dispatchers.Default) {
            for (i in 1..3) {
                Thread.sleep(100) // pretend we are computing it in CPU-consuming way
                emit(i) // emit next value
            }
        }
    }

    /**
     * 因为在main线程中调用collect，在flow中若是执行耗时操作则会明显地阻塞UI，此时需要开启新的后台线程来执行耗时操作，合理的切换Context的方法是通过flowOn方式
     */
    fun simpleFlowOn(): Flow<Int> = flow {
        for (i in 1..3) {
            Thread.sleep(100) // pretend we are computing it in CPU-consuming way
            log("Emitting $i")
            emit(i) // emit next value
        }
    }.flowOn(Dispatchers.Default) // RIGHT way to change context for CPU-consuming code in flow builder


    fun simpleComputation(): Flow<Int> = flow {
        for (i in 1..3) {
            delay(100) // pretend we are asynchronously waiting 100 ms
            emit(i) // emit next value
        }
    }

    /**
     * 当collect处理慢于flow内部执行，同时你只在乎最近一次的结果时，你可以选择conflate函数，比如flow内部已经发送了1、2、3，但是此时collect刚处理完1，那么现在他要处理的下一个值
     * 就是3，2就被舍弃了，这样你就可以减少collect的次数，从而减少collect的开销，从而提高性能。
     * 适用场景：比如你发送多个同样的请求，只需要用到最新的返回结果即可的时候
     */
    @Test
    fun test2() = runTest {
        simpleComputation()
            .conflate()
            .collect { value ->
                delay(300) // pretend we are processing it for 300 ms
                println(value)
            }
        //还有一种方式collectLatest获取最新的结果，他会在每次发送新值后取消当前collect块并启动新的collect块
        val time = measureTimeMillis {
            simpleComputation()
                .collectLatest { value -> // cancel & restart on the latest value
                    println("Collecting $value") //每次emit执行后这里都会执行到
                    //每次emit执行后这里都会被取消，直到最后一次
                    delay(300) // pretend we are processing it for 300 ms
                    println("Done $value")
                }
        }
        println("Collected in $time ms") //总时间大概是simpleComputation()的时间 + 最后一次collect的时间
    }

    /**
     * 合并多个flow
     */
    @Test
    fun test3() = runTest {
        val numbers = (1..3).asFlow().onEach { delay(300) } // numbers = 1, 2, 3
        val strings = flowOf('a', 'b', 'c').onEach { delay(400) }
        numbers.zip(strings) { a, b ->//a是前者的int，b是后者的char
            "$a -> $b"
        }.collect {
            println(it)
        }
        //combine用处：如果在两个flow的耗时不一样的时候，我们需要在collect中使用最新值来重新计算最新结果
        //比如，此时我们得到第一组数据“1 -> a”，第一个flow更快地产生了2，那么我们即刻会在collect中得到一个新的数据“2 -> a”
        numbers.combine(strings) { a, b ->//a是前者的int，b是后者的char
            "$a -> $b"
        }.collect {
            println(it)
        }
    }

    /**
     * flatten flows，当flow内部的元素也是flow时，可以通过flattenXxx系列函数变成一个一级flow
     */
    @OptIn(FlowPreview::class)
    @Test
    fun test4() = runTest {
        //result是一个Flow<Flow<T>>类型
//        val result = (1..3).asFlow().onEach { delay(100) }.map { requestFlow(it) }

        val startTime = System.currentTimeMillis()

        //flatMapConcat会等待前一个flow的执行完成，从而保证顺序性，这里也就是1时的requestFlow代码全部执行完成才会继续执行2的requestFlow
        //1: First at 27 ms from start
        //1: Second at 29 ms from start
        //2: First at 29 ms from start
        //2: Second at 29 ms from start
        //3: First at 29 ms from start
        //3: Second at 29 ms from start
        /*  (1..3).asFlow().onEach { delay(100) }.flatMapConcat { requestFlow(it) }.collect{
              println("$it at ${System.currentTimeMillis() - startTime} ms from start")
          }*/

        //flatMapMerge是flatMapConcat的变种，它不会等待前一个flow的emit完成，而是直接开始下一个flow的emit，从而提高性能（并发）
        //1: First at 45 ms from start
        //2: First at 49 ms from start
        //3: First at 50 ms from start
        //1: Second at 51 ms from start
        //2: Second at 51 ms from start
        //3: Second at 52 ms from start
        /*(1..3).asFlow().onEach { delay(100) } // a number every 100 ms
            .flatMapMerge { requestFlow(it) }
            .collect { value -> // collect and print
                println("$value at ${System.currentTimeMillis() - startTime} ms from start")
            }*/

        //和conflate、collectLatest一样，总是会拿当前最新的数据计算
        //1: First at 44 ms from start
        //2: First at 75 ms from start
        //3: First at 77 ms from start
        //3: Second at 78 ms from start
        (1..3).asFlow().onEach { delay(100) } // a number every 100 ms
            .flatMapLatest { requestFlow(it) }
            .collect { value -> // collect and print
                println("$value at ${System.currentTimeMillis() - startTime} ms from start")
            }

    }

    private fun requestFlow(i: Int): Flow<String> = flow {
        emit("$i: First")
        delay(500) // wait 500 ms
        emit("$i: Second")
    }

    /**
     * flow异常捕获方式
     */
    @Test
    fun test5() = runTest {
        //try-catch collect代码块，不管是在中间操作函数还是执行emit（这里是simple的flow）的函数中，都会被try-catch捕获，但是这会破坏异常透明性，因为不应该在所有使用处都添加try-catch
        /*try {
            simple().collect { value ->
                println(value)
                //value大于1则会抛出异常
                check(value <= 1) { "Collected $value" }
            }
        } catch (e: Throwable) {
            println("Caught $e")
        }*/

        //可以使用catch函数来捕获异常，并且它可以在捕获逻辑中再次emit一个元素进去，这样更灵活
        /*simple()
            .catch { e ->
                println("Caught $e")
                emit(-1)
            }
            .collect { value ->
                println(value)
                //但是catch无法捕获它下游的异常，比如这里的异常仍然会抛出
                check(value <= 1) { "Collected $value" }
            }*/

        //为了解决catch无法捕获它下游的异常的问题，我们就不在collect中写代码
        simple()
            .onEach { value ->
                println(value)
                //但是catch无法捕获它下游的异常，比如这里的异常仍然会抛出
                check(value <= 1) { "Collected $value" }
            }
            .catch { e ->
                println("Caught $e")
                emit(-1) //不过这里再发送也没有下游逻辑处理了
            }
            .collect()
    }

    fun simple(): Flow<Int> = flow {
        for (i in 1..3) {
            println("Emitting $i")
            emit(i) // emit next value
        }
    }.map {
        check(it <= 1) { "Crashed on $it" }
        it
    }

    /**
     * flow结束时的操作（正常结束或者异常结束）
     * 一种是命令式，一种是声明式
     */
    @Test
    fun test6() = runTest {
        //命令式
        /*try {
            simple().collect { value -> println(value) }
        } finally {
            println("Done")
        }*/
        //声明式
        simpleWithError()
            .catch { cause -> println("Caught exception") } //catch如果定义在前面，后面的所有onCompletion语句都不会再拿到异常对象，而后面的catch语句也不会再执行
            //onCompletion可以拿到一个异常对象（如果上游抛出异常或者取消的话）
            .onCompletion { cause -> if (cause != null) println("Flow completed exceptionally") else println("Flow completed normally") }
            //onCompletion中的异常对象还是会继续往下传递,因此下面还是会输出Flow completed exceptionally 2~~
            .onCompletion { cause -> if (cause != null) println("Flow completed exceptionally 2~~") else println("Flow completed normally 2~~") }
            //onCompletion不会处理异常，所以如果你没加catch的话会抛出异常（如果有异常的话）
            .catch { cause -> println("Caught exception2") }
            .collect { value -> println(value) }
    }

    fun simpleWithError(): Flow<Int> = flow {
        emit(1)
        throw RuntimeException()

    }

    /**
     * 给flow添加监听器
     */
    @Test
    fun test7() = runTest {
        events()
            .onEach { event -> println("Event: $event") }  //onEach可以充当listener的角色
//            .collect() //collect的方式是同步串行的，因此监听的过程必须要在当前线程等待
            .launchIn(this) // launch可以将flow的收集操作放到一个协程中，这样就变成异步，不会影响当前线程后面的执行，当前线程被强制结束时该flow也会被关闭（即使没结束）
//            .launchIn(Dispatchers.Default)
        println("Done")
    }

    fun events(): Flow<Int> = (1..3).asFlow().onEach { delay(100) }

    /**
     * flow的取消检测
     */
    @Test
    fun test8() = runTest {
        //对于flow{}操作符来说，cancel是正常奏效的，注意，取消的是collect的收集，但是在取消完成之前，flow的上游的代码还是在执行，只是不再接收，比如这里的demo在cancel之后foo函数内还是会打印出“Emitting 4”
        /*foo().collect { value ->
            if (value == 3) cancel() //cancel是存在于CoroutineScope中的函数，所以需要在一个协程环境中才能使用
            println(value)
        }*/
        //对于其他大部分操作符都不支持即时cancel，除非添加.onEach { currentCoroutineContext().ensureActive() }或者.cancellable()
        (1..5).asFlow()
//            .onEach { currentCoroutineContext().ensureActive() }
            .cancellable()
            .collect { value ->
                if (value == 3) cancel() //asFlow()的话这里不会被取消，1到5都会打印出来
                println(value)
            }
    }

    fun foo(): Flow<Int> = flow {
        for (i in 1..5) {
            println("Emitting $i")
            emit(i)
        }
    }

}